// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.core.tsuid;

import com.stumbleupon.async.Deferred;
import net.opentsdb.core.IncomingDataPoints;
import net.opentsdb.core.TSDB;
import net.opentsdb.core.TsuidExchange;
import net.opentsdb.plugin.impl.LocalMetaCache;
import net.opentsdb.uid.UniqueId;
import net.opentsdb.utils.Config;
import net.opentsdb.utils.enums.TsuidChecks;
import net.opentsdb.utils.expiringmap.ExpiringMap;
import net.opentsdb.utils.hbase.HbaseMetaUtils;
import net.opentsdb.utils.hbase.HbaseWrapper;
import org.hbase.async.Bytes;
import org.hbase.async.HBaseClient;
import org.hbase.async.KeyValue;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Matchers;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import org.powermock.reflect.Whitebox;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.*;
import static org.powermock.api.mockito.PowerMockito.mock;

/**
 * tsuid 状态以及加载测试
 */
@RunWith(PowerMockRunner.class)
@PrepareForTest({ TSDB.class, Config.class, UniqueId.class, HBaseClient.class,
        IncomingDataPoints.class })
public class TestTsuidCount{

    private static Config config;
    private static TSDB tsdb = null;
    private HBaseClient client = mock(HBaseClient.class);
    private UniqueId metrics = mock(UniqueId.class);
    private UniqueId tag_names = mock(UniqueId.class);
    private UniqueId tag_values = mock(UniqueId.class);
    private HbaseMetaUtils metaUtils = mock(HbaseMetaUtils.class);
    private Map tags;

    @SuppressWarnings("unchecked")
    @Before
    public void before() throws Exception {
        config = new Config(false);
        config.overrideConfig("tsd.http.service", "127.0.0.1:10010");
        config.overrideConfig("tsd.http.service.path", "/tsdb/v1/dashboard/tsdbmeta/receiveMeta");
        config.overrideConfig("tsd.worker.id", "101");
        config.overrideConfig("tsd.core.meta.cache.enable", "true");
        config.overrideConfig("tsd.timeseriesfilter.enable", "true");
        config.overrideConfig("tsd.core.meta.tsuid.max_entries", "2");

        tsdb = new TSDB(client, config);

        // replace the "real" field objects with mocks
        Field met = tsdb.getClass().getDeclaredField("metrics");
        met.setAccessible(true);
        met.set(tsdb, metrics);

        Field tagk = tsdb.getClass().getDeclaredField("tag_names");
        tagk.setAccessible(true);
        tagk.set(tsdb, tag_names);

        Field tagv = tsdb.getClass().getDeclaredField("tag_values");
        tagv.setAccessible(true);
        tagv.set(tsdb, tag_values);

        when(metrics.width()).thenReturn((short) 3);
        when(tag_names.width()).thenReturn((short) 3);
        when(tag_values.width()).thenReturn((short) 3);

        when(metrics.getId("foo")).thenReturn(new byte[] { 0, 0, 1 });
        when(metrics.getOrCreateId("foo")).thenReturn(new byte[] { 0, 0, 1 });
        when(metrics.getId("metric")).thenReturn(new byte[] { 0, 0, 2 });
        when(metrics.getOrCreateId("metric")).thenReturn(new byte[] { 0, 0, 2 });
        when(metrics.getId("metric_3")).thenReturn(new byte[] { 0, 0, 3 });
        when(metrics.getOrCreateId("metric_3")).thenReturn(new byte[] { 0, 0, 3 });
        when(tag_names.getId("testk")).thenReturn(new byte[] { 0, 0, 1 });
        when(tag_names.getOrCreateId("testk")).thenReturn(new byte[] { 0, 0, 1 });
        when(tag_values.getId("testv")).thenReturn(new byte[] { 0, 0, 1 });
        when(tag_values.getOrCreateId("testv")).thenReturn(new byte[] { 0, 0, 1 });
        when(metrics.getNameAsync(new byte[] { 0, 0, 1 })).thenReturn(
                Deferred.fromResult("foo"));

        PowerMockito.mockStatic(IncomingDataPoints.class);
        final byte[] row = new byte[] { 0, 0, 1, 0, 0, 0, 0, 0, 0, 1, 0, 0, 1 };
        final byte[] row3 = new byte[] { 0, 0, 3, 0, 0, 0, 0, 0, 0, 1, 0, 0, 1 };
        PowerMockito.doAnswer(new Answer<byte[]>() {
            public byte[] answer(final InvocationOnMock unused) throws Exception {
                return row;
            }
        }).when(IncomingDataPoints.class, "rowKeyTemplate", (TSDB) any(),
                anyString(), (Map<String, String>) any());

        PowerMockito.doAnswer(new Answer<byte[]>() {
            public byte[] answer(final InvocationOnMock unused) throws Exception {
                return row3;
            }
        }).when(IncomingDataPoints.class, "rowKeyTemplate", (TSDB) any(),
                Matchers.eq("metric_3"), (Map<String, String>) any());

        HbaseWrapper hbaseWrapper = tsdb.getHbaseWrapper();

        Field utils = hbaseWrapper.getClass().getDeclaredField("metaUtils");
        utils.setAccessible(true);
        utils.set(hbaseWrapper, metaUtils);

        when(metaUtils.scanTableData(any())).thenReturn(new ArrayList<>());
        when(metaUtils.fetchData(any(), any(), any())).thenReturn(Collections.EMPTY_LIST);
        when(metaUtils.existsTsuid(any())).thenReturn(true);

        tsdb.initializePlugins(true);

        tags = new HashMap<String,String>();
        tags.put("testk","testv");
        tsdb.addPoint("foo",System.currentTimeMillis(),0L,tags);
    }

    @Test
    public void metaActor(){
        assertNotNull("meta操作信息",tsdb.getHbaseWrapper().metaUtils);
    }

    @Test
    public void tagActor(){
        assertNotNull("tag操作信息",tsdb.getHbaseWrapper().tagUtils);
    }

    @Test
    public void checkLocalExistsStatus(){
        byte[] tsuid = TsuidExchange.getTsuidFromData(tsdb, "foo", tags);

        assertTrue("检查本地存在tsuid状态",
                tsdb.getMetaCache().checks(tsuid) ==
                        TsuidChecks.LocalTsuidExists);
    }

    @Test
    public void checkHbaseExistsStatus(){
        byte[] notTsuid = TsuidExchange.getTsuidFromData(tsdb, "metric", tags);

        assertTrue("检查本地不存在状态, hbase存在",
                tsdb.getMetaCache().checks(notTsuid) ==
                        TsuidChecks.HbaseTsuidExists);
    }

    @Test
    public void checkLocalCountFullStatus(){
        byte[] notTsuid = TsuidExchange.getTsuidFromData(tsdb, "metric", tags);

        tsdb.addPoint("metric_3",System.currentTimeMillis(),0L,tags);
        when(metaUtils.existsTsuid(notTsuid)).thenReturn(false);
        assertTrue("检查本地不存在状态, hbase不存在，本地计数已满",
                tsdb.getMetaCache().checks(notTsuid) ==
                        TsuidChecks.LocalCountFullCapacity);
    }

    @Test
    public void checkHbaseCountFullStatus(){
        byte[] notTsuid = TsuidExchange.getTsuidFromData(tsdb, "metric", tags);
        when(metaUtils.existsTsuid(notTsuid)).thenReturn(false);
        when(metaUtils.getTsuidCount()).thenReturn(2L);

        assertTrue("检查本地不存在状态, hbase不存在，本地计数未满，hbase计数已满",
                tsdb.getMetaCache().checks(notTsuid) ==
                        TsuidChecks.HbaseCountFullCapacity);
    }

    @Test
    public void checkNewDataStatus(){
        byte[] notTsuid = TsuidExchange.getTsuidFromData(tsdb, "metric", tags);
        when(metaUtils.existsTsuid(notTsuid)).thenReturn(false);
        when(metaUtils.getTsuidCount()).thenReturn(1L);

        assertTrue("通过所有检查，新数据",
                tsdb.getMetaCache().checks(notTsuid) ==
                        TsuidChecks.NewInputTsuid);
    }

    @Test
    public void loadTsuidTest(){
        when(metrics.getId("metric1")).thenReturn(new byte[] { 0, 0, 5 });
        when(metrics.getId("metric2")).thenReturn(new byte[] { 0, 0, 6 });
        when(metrics.getOrCreateId("metric1")).thenReturn(new byte[] { 0, 0, 5 });
        when(metrics.getOrCreateId("metric2")).thenReturn(new byte[] { 0, 0, 6 });

        LocalMetaCache cache = (LocalMetaCache)tsdb.getMetaCache();
        ArrayList<ArrayList<KeyValue>> results = new ArrayList<>();
        byte[] tsuid1 = TsuidExchange.getTsuidFromData(tsdb, "metric1", tags);
        byte[] tsuid2 = TsuidExchange.getTsuidFromData(tsdb, "metric2", tags);
        byte[] family = new byte[]{0,0,1};
        byte[] qulitifer = new byte[]{0,0,1};

        KeyValue k1 = new KeyValue(tsuid1,family,qulitifer, Bytes.fromLong(0));
        KeyValue k2 = new KeyValue(tsuid2,family,qulitifer, Bytes.fromLong(0));
        ArrayList<KeyValue> list = new ArrayList<>();
        list.add(k1);
        list.add(k2);
        results.add(list);

        when(metaUtils.scanTableData(any())).thenReturn(results);

        /*
        mock ExpiringMap 执行 verify 判断
        ExpiringMap map = mock(ExpiringMap.class);
        // 替换metaMap
        Whitebox.setInternalState(cache,"metaMap",map);
        // 执行loadMeta方法
        try{
            Whitebox.invokeMethod(cache,"loadMeta",tsdb);
        }catch (Exception e){
            e.printStackTrace();
        }
        verify(map, times(2)).put(any(),any(),anyLong(),anyLong());
        */

        when(metaUtils.getTsuidCount()).thenReturn(null);
        cache.initialize(tsdb);

        ExpiringMap map = Whitebox.getInternalState(cache, "metaMap");

        assertTrue("加载hbase两条数据",map.size() == 2);
        verify(metaUtils, times(1)).initTSMetaCount(2L);
    }

}
